/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package generator

import java.io.File
import java.io.FileOutputStream

import scala.collection.JavaConversions._

import com.google.protobuf.ByteString
import com.google.protobuf.CodedOutputStream

import aggr.SerializedTypes.SerializedType
import aggr.Aggregators

import pbdata.Tables.KeyValue

/**
 * Base trait of emitter.
 *
 * Subclass should handle key and value with overriding write() method.
 */
trait Emitter {

  /**
   * Emits key/value date.
   * This function is called when "emit statement" of Sawzall Clone is executed.
   */
  final def emit(table: Helpers.Table, key: Helpers.ByteStringWrapper, value: Helpers.ByteStringWrapper): Unit = {
    write(table.id, key.get, value.get)
  }

  /**
   * This function actually write emitted date. 
   */
  def write(id: Int, key: ByteString, value: ByteString): Unit
}

/**
 * Emitter for local execution.
 */
class LocalEmitter(typeMap: Map[String, SerializedType], tables: List[TableValue]) extends Emitter {

  /**
   * Emitter for each table.
   */
  private class TableEmitter(table: TableValue) {
    val tmp = File.createTempFile("scdata", null)
    val os  = new FileOutputStream(tmp)
    val cos = CodedOutputStream.newInstance(os)
    var process: Option[Aggregators.Process] = None

    def write(key: ByteString, value: ByteString): Unit = {
      cos.writeMessage(1, KeyValue.newBuilder().setKey(key).setValue(value).build)
    }

    /**
     * Close temporary file.
     */
    def close(): Unit = {
      cos.flush()
      os.close()
    }

    /**
     * Invokes aggregator.
     */
    def invoke(outputText: Boolean): Unit = {
      close()

      val keyType = typeMap(table.keyType)
      val valType = typeMap(table.valType)

      val p = Aggregators.kick(
        "%03d-%s.table".format(table.id, table.name),
        table.aggr, 
        keyType, 
        valType, 
        table.args.map{case x: Long => x}, tmp.toString(), outputText)

      process = Some(p)
    }

    /**
     * Waits Aggregator to finish.
     */
    def waitFor(): Int = {
      try {
        process.map{_.waitFor()}.getOrElse(1)
      } finally {
        process = None
        tmp.delete()
      }
    }
  }

  private val tableEmitters: Map[Int, TableEmitter] = Map() ++ tables.map { table => 
    (table.id, new TableEmitter(table))
  }

  def write(tableId: Int, key: ByteString, value: ByteString): Unit = {
    tableEmitters(tableId).write(key, value)
  }

  /**
   * Invokes aggregators.
   */
  def invokeTables(outputText: Boolean): Int = {
    for (table <- tableEmitters.values) {
      table.invoke(outputText)
    }
    return tableEmitters.values.map(x => x.waitFor()).filter(_ != 0).size
  }
}
